--[[

    Filename:	auxthr.lua

    Programmer:	aesiot

    Remark:

--]]

local auxlib = require( "auxlib" )

local auxthr = { _VERSION = "1.0" }

local _ThrReg = {}
local _ThrTag = {}

--------------------------------------------------------------------------------

function auxthr.create( func, atexit, name )

	auxlib.argcheck( auxlib.is_nil( atexit ) or auxlib.is_function( atexit ), 2, "(function expected)" )

	auxlib.argcheck( auxlib.is_nil( name ) or auxlib.is_string( name ), 3, "(string expected)" )

	local co = coroutine.create( func )

	_ThrReg[ co ] = { name = name, atexit = atexit }

	return co
end

--------------------------------------------------------------------------------

function auxthr.resume( co, tag, ... )

	local registry = _ThrReg[ co ]

	if not registry then

		return coroutine.resume( co, ... )
	end

	if tag ~= registry.tag then

		error( "coroutine code error. tag mismatch", 2 )
	end

	local result = { coroutine.resume( co, ... ) }

	if "dead" == coroutine.status( co ) then

		_ThrReg[ co ] = nil

		if registry.atexit then

			auxlib.pcall( registry.atexit, table.unpack( result ) )

		elseif ( not result[ 1 ] ) and auxlib.is_string( result[ 2 ] ) then

			logger.print_multi(

				logger.LEVEL_WARNING, "[ WARN ] LUA RUNTIME ERROR",

				logger.LEVEL_DEBUG, string.format( "  # %s", result[ 2 ] )
			)
		end

		return table.unpack( result )
	end

	if result[ 2 ] ~= _ThrTag then

		error( "coroutine code error. yield no tag", 2 )
	end

	registry.tag = _ThrTag.tag

	return true, table.unpack( result, 3 )
end

--------------------------------------------------------------------------------

function auxthr.yield( tag, ... )

	local co = coroutine.running()

	if _ThrReg[ co ] then

		auxlib.argcheck( auxlib.is_string( tag ), 1, "(string expected)" )

		_ThrTag.tag = tag

		return coroutine.yield( _ThrTag, ... )
	else

		return coroutine.yield( ... )
	end
end

--------------------------------------------------------------------------------

function auxthr.get_name( co )

	if nil == co then

		co = coroutine.running()
	end

	for key, val in pairs( _ThrReg ) do

		if co == key then

			return val.name
		end
	end

	return nil
end

--------------------------------------------------------------------------------

function auxthr.get_thread( name )

	for key, val in pairs( _ThrReg ) do

		if val.name == name then

			return key
		end
	end

	return nil
end

--------------------------------------------------------------------------------

function auxthr.run( func, atexit, name, ... )

	return auxthr.resume( auxthr.create( func, atexit, name ), nil, ... )
end

--------------------------------------------------------------------------------

function auxthr.send( vmname, method, ... )

	local tag = tostring( os.get_tick_count() )

	local auxmsg = auxlib.assert( require( "auxmsg" ) )

	local thread, ismain = coroutine.running()

	auxlib.assert( not ismain, "attempt to yield from outside a coroutine" )

	auxmsg.send( vmname, function( result )

		auxthr.resume( thread, tag, result )

	end, method, ... )

	return auxthr.yield( tag )
end

--------------------------------------------------------------------------------

function auxthr.sleep( msec )

	local tag = tostring( os.get_tick_count() )

	local auxtmr = auxlib.assert( require( "auxtmr" ) )

	local thread, ismain = coroutine.running()

	auxlib.assert( not ismain, "attempt to yield from outside a coroutine" )

	auxtmr.create( nil, msec, 0, function()

		auxthr.resume( thread, tag, msec )
	end )

	return auxthr.yield( tag )
end

--------------------------------------------------------------------------------

function auxthr.import( vmname, method )

	auxlib.argcheck( auxlib.is_string( vmname ) or auxlib.is_nil( vmname ), 1, "(string expected)" )

	auxlib.argcheck( auxlib.is_string( method ), 2, "(string expected)" )

	return function( ... )

		return auxthr.send( vmname, method, ... )
	end
end

--------------------------------------------------------------------------------

function auxthr.mqtt_rpc_create( conf, timeout )

	local auxtmr = auxlib.assert( require( "auxtmr" ) )
	local auxrpc = auxlib.assert( require( "auxrpc" ) )

	local tag = tostring( os.get_tick_count() )

	local thread, ismain = coroutine.running()

	auxlib.assert( not ismain, "attempt to yield from outside a coroutine" )

	local result = auxrpc.mqtt_rpc_create( conf )

	result.on_ready = function( self )

		self.on_close     = nil
		self.on_ready     = nil
		self.on_reconnect = nil

		auxthr.resume( thread, tag, self )
	end

	result.on_close = function( self )

		self.on_close     = nil
		self.on_ready     = nil
		self.on_reconnect = nil

		auxthr.resume( thread, tag, nil )
	end

	result.on_reconnect = function( self )

		self.on_close     = nil
		self.on_ready     = nil
		self.on_reconnect = nil

		auxthr.resume( thread, tag, nil )
	end

	if auxlib.is_integer( timeout ) then

		auxtmr.create( nil, timeout, 0, function()

			pcall( result.close, result )

			result.on_close     = nil
			result.on_ready     = nil
			result.on_reconnect = nil

			result = nil

			auxthr.resume( thread, tag, nil )
		end )
	end

	return auxthr.yield( tag )
end

--------------------------------------------------------------------------------

function auxthr.udp_rpc_create( addr )

	local auxrpc = auxlib.assert( require( "auxrpc" ) )

	local tag = tostring( os.get_tick_count() )

	local thread, ismain = coroutine.running()

	auxlib.assert( not ismain, "attempt to yield from outside a coroutine" )

	local result = auxrpc.udp_rpc_create( addr )

	result.on_ready = function( self )

		self.on_close = nil
		self.on_ready = nil

		auxthr.resume( thread, tag, self )
	end

	result.on_close = function( self )

		self.on_close = nil
		self.on_ready = nil

		auxthr.resume( thread, tag, nil )
	end

	return auxthr.yield( tag )
end

--------------------------------------------------------------------------------

function auxthr.rpc2sync( func )

	return function( ... )

		local co = coroutine.running()

		local tag = tostring( os.get_tick_count() )

		local b, r = func( function( r, c ) auxthr.resume( co, tag, r, c ) end, ... )

		if not b then

			return nil, nil
		else

			return auxthr.yield( tag )
		end
	end
end

--------------------------------------------------------------------------------

function auxthr.mqtt_tcp_create( host, port, username, password, clientid, timedout, keep_alive )

	local MQTT    = auxlib.assert( require( "MQTT"    ) )
	local auxmqtt = auxlib.assert( require( "auxmqtt" ) )

	local result = nil

	do
		local tag = tostring( os.get_tick_count() )

		local thread, ismain = coroutine.running()

		auxlib.assert( not ismain, "attempt to yield from outside a coroutine" )

		result = auxmqtt.new( host, port )

		local timeid = auxtmr.create( nil, timedout or 60000, 0, function()

			result:close()
		end )

		function result:on_ready()

			self:login( keep_alive or 120, clientid, username, password )
		end

		function result:on_connected()

			self.on_ready     = nil
			self.on_session   = nil
			self.on_connected = nil
			self.on_reconnect = nil

			pcall( auxtmr.remove, timeid )

			auxthr.resume( thread, tag, true )
		end

		function result:on_session( ident, errno )

			self.on_ready     = nil
			self.on_session   = nil
			self.on_connected = nil
			self.on_reconnect = nil

			pcall( auxtmr.remove, timeid )

			auxthr.resume( thread, tag, false, string.format( "Login server failed (%d)", errno ) )
		end

		function result:on_reconnect()

			self.on_ready     = nil
			self.on_session   = nil
			self.on_connected = nil
			self.on_reconnect = nil

			pcall( auxtmr.remove, timeid )

			auxthr.resume( thread, tag, false, "Unable to connect to server" )
		end

		auxlib.assert( auxthr.yield( tag ) )
	end

	local observe     = auxlib.bind1st( result, result.observe     )
	local publish     = auxlib.bind1st( result, result.publish     )
	local subscribe   = auxlib.bind1st( result, result.subscribe   )
	local unsubscribe = auxlib.bind1st( result, result.unsubscribe )

	function result:observe( topic, qos, proc )

		local tag = tostring( os.get_tick_count() )

		local thread, ismain = coroutine.running()

		auxlib.assert( not ismain, "attempt to yield from outside a coroutine" )

		subscribe( topic, qos, function( self, code )

			auxthr.resume( thread, tag, code )
		end )

		local code = auxthr.yield( tag )

		if 0x80 ~= code then

			self:bind( topic, proc )
		end

		return code
	end

	function result:publish( topic, message, qos, retain )

		local tag = tostring( os.get_tick_count() )

		local thread, ismain = coroutine.running()

		auxlib.assert( not ismain, "attempt to yield from outside a coroutine" )

		if auxlib.is_none( qos ) or qos == MQTT.QoS0 then

			publish( topic, message, qos, retain )

			return 0
		else

			publish( topic, message, qos, retain, function( self, code )

				auxthr.resume( thread, tag, code )
			end )

			return auxthr.yield( tag )
		end
	end

	function result:subscribe( topic, qos )

		local tag = tostring( os.get_tick_count() )

		local thread, ismain = coroutine.running()

		auxlib.assert( not ismain, "attempt to yield from outside a coroutine" )

		subscribe( topic, qos, function( self, code )

			auxthr.resume( thread, tag, code )
		end )

		return auxthr.yield( tag )
	end

	function result:unsubscribe( topic )

		local tag = tostring( os.get_tick_count() )

		local thread, ismain = coroutine.running()

		auxlib.assert( not ismain, "attempt to yield from outside a coroutine" )

		unsubscribe( topic, function( self, code )

			auxthr.resume( thread, tag, code )
		end )

		return auxthr.yield( tag )
	end

	return result
end

--------------------------------------------------------------------------------

return auxthr
